package com.yin.myproject.zk;

import com.google.gson.*;
import com.yin.myproject.DynamicTaskManager;
import com.yin.myproject.core.IScheduleDataManager;
import com.yin.myproject.core.ScheduleServer;
import com.yin.myproject.core.ScheduleTask;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Type;
import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author Eason
 * @create 2018-04-08 14:13
 **/
public class ScheduleDataManager4ZK implements IScheduleDataManager {
    private static final transient Logger LOG = LoggerFactory.getLogger(ScheduleDataManager4ZK.class);

    private static final String NODE_SERVER = "server";
    private static final String NODE_TASK = "task";
    private static final String NODE_HISTORY = "history";
    private static final long SERVER_EXPIRE_TIME = 15 * 1000;
    private Gson gson;
    private ZKManager zkManager;
    private String pathServer;
    private String pathTask;
    private String pathHistory;
    private long zkBaseTime = 0;
    private long loclaBaseTime = 0;
    private Random random;

    public ScheduleDataManager4ZK(ZKManager aZkManager) throws Exception {
        this.zkManager = aZkManager;
        gson = new GsonBuilder().registerTypeAdapter(Timestamp.class, new TimestampTypeAdapter())
                .setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        this.pathServer = this.zkManager.getRootPath() + "/" + NODE_SERVER;
        this.pathTask = this.zkManager.getRootPath() + "/" + NODE_TASK;
        this.pathHistory = this.zkManager.getRootPath() + "/" + NODE_HISTORY;
        this.random = new Random();
        if (this.getZooKeeper().exists(this.pathServer, false) == null) {
            ZKTools.createPath(getZooKeeper(), this.pathServer, CreateMode.PERSISTENT, this.zkManager.getAcl());
        }
        loclaBaseTime = System.currentTimeMillis();

        String tempPath = this.zkManager.getZooKeeper().create(this.zkManager.getRootPath() + "/systime", null,
                this.zkManager.getAcl(), CreateMode.EPHEMERAL_SEQUENTIAL);
        Stat tempStat = this.zkManager.getZooKeeper().exists(tempPath, false);
        zkBaseTime = tempStat.getCtime();
        ZKTools.deleteTree(getZooKeeper(), tempPath);
        if (Math.abs(this.zkBaseTime - this.loclaBaseTime) > 5000) {
            LOG.error("请注意，Zookeeper服务器时间与本地时间相差 ： " + Math.abs(this.zkBaseTime - this.loclaBaseTime) + " ms");
        }
    }

    public ZooKeeper getZooKeeper() throws Exception {
        return this.zkManager.getZooKeeper();
    }

    /**
     * 发送心跳信息
     *
     * @param server
     * @throws Exception
     */
    @Override
    public boolean refreshScheduleServer(ScheduleServer server) throws Exception {
        Timestamp heartBeatTime = new Timestamp(this.getSystemTime());
        String zkPath = this.pathServer + "/" + server.getUuid();
        if (this.getZooKeeper().exists(zkPath, false) == null) {
            // 数据可能被清除，先清除内存数据后，重新注册数据
            server.setRegister(false);
            return false;
        } else {
            Timestamp oldHeartBeatTime = server.getHeartBeatTime();
            server.setHeartBeatTime(heartBeatTime);
            server.setVersion(server.getVersion() + 1);
            String valueString = this.gson.toJson(server);
            try {
                byte[] bytes = ZKTools.serialize(valueString);
                this.getZooKeeper().setData(zkPath, bytes, -1);
                // this.getZooKeeper().setData(zkPath, valueString.getBytes(), -1);
            } catch (Exception e) {
                // 恢复上次的心跳时间
                server.setHeartBeatTime(oldHeartBeatTime);
                server.setVersion(server.getVersion() - 1);
                throw e;
            }
            return true;
        }
    }

    /**
     * 注册服务器，如果之前已经注册过，且
     *
     * @param server
     * @throws Exception
     */
    @Override
    public void registerScheduleServer(ScheduleServer server) throws Exception {
        if (server.isRegister()) {
            throw new Exception(server.getUuid() + " 被重复注册");
        }
        // clearExpireScheduleServer();
        String realPath;
        // 此处必须增加UUID作为唯一性保障
        StringBuffer id = new StringBuffer();
        id.append(server.getIp()).append("$").append(UUID.randomUUID().toString().replaceAll("-", "").toUpperCase());
        String zkServerPath = pathServer + "/" + id.toString() + "$";
        realPath = this.getZooKeeper().create(zkServerPath, null, this.zkManager.getAcl(),
                CreateMode.PERSISTENT_SEQUENTIAL);
        server.setUuid(realPath.substring(realPath.lastIndexOf("/") + 1));

        Timestamp heartBeatTime = new Timestamp(getSystemTime());
        server.setHeartBeatTime(heartBeatTime);

        String valueString = this.gson.toJson(server);
        byte[] bytes = ZKTools.serialize(valueString);
        // this.getZooKeeper().setData(realPath, valueString.getBytes(), -1);
        this.getZooKeeper().setData(realPath, bytes, -1);
        server.setRegister(true);
    }

    /**
     * 加载所有调度服务节点
     *
     * @return
     * @throws Exception
     */
    public List<String> loadAllScheduleServer() throws Exception {
        String zkPath = this.pathServer;
        List<String> names = this.getZooKeeper().getChildren(zkPath, false);
        Collections.sort(names);
        return names;
    }

    /**
     * 清理过期调度服务节点
     *
     * @throws Exception
     */
    @Override
    public void clearExpireScheduleServer() throws Exception {
        String zkPath = this.pathServer;
        if (this.getZooKeeper().exists(zkPath, false) == null) {
            this.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
        }
        List<String> serverNames = this.zkManager.getZooKeeper().getChildren(zkPath, false);
        for (String name : serverNames) {
            try {
                Stat stat = new Stat();
                this.getZooKeeper().getData(zkPath + "/" + name, null, stat);
                if (getSystemTime() - stat.getMtime() > SERVER_EXPIRE_TIME) {
                    ZKTools.deleteTree(this.getZooKeeper(), zkPath + "/" + name);
                    LOG.debug("ScheduleServer[" + zkPath + "/" + name + "]过期清除");
                }
            } catch (Exception e) {
                // 当有多台服务器时，存在并发清理的可能，忽略异常
            }
        }
    }

    /**
     * 注销调度服务节点
     *
     * @param server
     * @throws Exception
     */
    @Override
    public void unRegisterScheduleServer(ScheduleServer server) throws Exception {
        List<String> serverList = this.loadScheduleServerNames();

        if (server.isRegister() && this.isLeader(server.getUuid(), serverList)) {
            // delete task
            String zkPath = this.pathTask;
            String serverPath = this.pathServer;

            if (this.getZooKeeper().exists(zkPath, false) == null) {
                this.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
            }

            // get all task
            List<String> children = this.getZooKeeper().getChildren(zkPath, false);
            if (null != children && children.size() > 0) {
                for (String taskName : children) {
                    String taskPath = zkPath + "/" + taskName;
                    if (this.getZooKeeper().exists(taskPath, false) != null) {
                        ZKTools.deleteTree(this.getZooKeeper(), taskPath + "/" + server.getUuid());
                    }
                }
            }

            // 删除
            if (this.getZooKeeper().exists(this.pathServer, false) == null) {
                ZKTools.deleteTree(this.getZooKeeper(), serverPath + serverPath + "/" + server.getUuid());
            }
            server.setRegister(false);
        }
    }

    /**
     * 查询调度服务节点名称集合
     *
     * @throws Exception
     */
    @Override
    public List<String> loadScheduleServerNames() throws Exception {
        String zkPath = this.pathServer;
        if (this.getZooKeeper().exists(zkPath, false) == null) {
            return new ArrayList<String>();
        }
        List<String> serverList = this.getZooKeeper().getChildren(zkPath, false);
        Collections.sort(serverList, new Comparator<String>() {
            public int compare(String u1, String u2) {
                return u1.substring(u1.lastIndexOf("$") + 1).compareTo(u2.substring(u2.lastIndexOf("$") + 1));
            }
        });
        return serverList;
    }

    /**
     * 分配任务
     *
     * @param currentUuid    调度服务标识
     * @param taskServerList 调度服务节点列表
     */
    @Override
    public void assignTask(String currentUuid, List<String> taskServerList) throws Exception {
        if (!this.isLeader(currentUuid, taskServerList)) {
            LOG.debug(currentUuid + ":不是负责任务分配的Leader,直接返回");
            return;
        }
        LOG.debug(currentUuid + ":开始重新分配任务......");
        if (taskServerList.isEmpty()) {
            // 在服务器动态调整的时候，可能出现服务器列表为空的清空
            return;
        }
        if (this.zkManager.checkZookeeperState()) {
            String zkPath = this.pathTask;
            if (this.getZooKeeper().exists(zkPath, false) == null) {
                this.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
            }
            List<String> children = this.getZooKeeper().getChildren(zkPath, false);
            if (null != children && !children.isEmpty()) {
                for (String taskName : children) {
                    String taskPath = zkPath + "/" + taskName;
                    if (this.getZooKeeper().exists(taskPath, false) == null) {
                        this.getZooKeeper().create(taskPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
                    }

                    List<String> taskServerIds = this.getZooKeeper().getChildren(taskPath, false);
                    if (null == taskServerIds || taskServerIds.isEmpty()) {
                        assignServer2Task(taskServerList, taskPath);
                    } else {
                        boolean hasAssignSuccess = false;
                        for (String serverId : taskServerIds) {
                            if (taskServerList.contains(serverId)) {
                                hasAssignSuccess = true;
                                continue;
                            }
                            ZKTools.deleteTree(this.getZooKeeper(), taskPath + "/" + serverId);
                        }
                        if (!hasAssignSuccess) {
                            assignServer2Task(taskServerList, taskPath);
                        }
                    }

                }
            } else {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(currentUuid + ":没有集群任务");
                }
            }
        }

    }

    /**
     * 重新分配任务
     *
     * @param taskServerList
     * @param taskPath
     * @throws Exception
     */
    private void assignServer2Task(List<String> taskServerList, String taskPath) throws Exception {
        int index = random.nextInt(taskServerList.size());
        String serverId = taskServerList.get(index);
        LOG.info("assign task to server, the taskPath is {}, the serverId is {}", taskPath, serverId);
        this.getZooKeeper().create(taskPath + "/" + serverId, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Assign server [{}]" + " to task [{}]", serverId, taskPath);
        }
    }

    /**
     * 判断是否为主服务
     *
     * @param uuid       调度服务id
     * @param serverList 服务集群列表
     */
    @Override
    public boolean isLeader(String uuid, List<String> serverList) {
        return uuid.equals(getLeader(serverList));
    }

    /**
     * 获取主服务标识
     *
     * @param serverList
     * @return
     */
    private String getLeader(List<String> serverList) {
        if (serverList == null || serverList.isEmpty()) {
            return "";
        }
        LOG.debug("get Leader Server: {}", serverList);
        long no = Long.MAX_VALUE;
        long tmpNo = -1;
        String leader = null;
        for (String server : serverList) {
            LOG.debug("server : {}", server);
            tmpNo = Long.parseLong(server.substring(server.lastIndexOf("$") + 1), 16);
            if (no > tmpNo) {
                no = tmpNo;
                leader = server;
            }
        }
        LOG.debug("leader: {}", leader);
        return leader;
    }

    private long getSystemTime() {
        return this.zkBaseTime + (System.currentTimeMillis() - this.loclaBaseTime);
    }

    private class TimestampTypeAdapter implements JsonSerializer<Timestamp>, JsonDeserializer<Timestamp> {

        @Override
        public JsonElement serialize(Timestamp src, Type arg1, JsonSerializationContext arg2) {
            DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String dateFormatAsString = format.format(new Date(src.getTime()));
            return new JsonPrimitive(dateFormatAsString);
        }

        @Override
        public Timestamp deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) {
            if (!(json instanceof JsonPrimitive)) {
                throw new JsonParseException("The date should be a string value");
            }

            try {
                DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date date = format.parse(json.getAsString());
                return new Timestamp(date.getTime());
            } catch (Exception e) {
                throw new JsonParseException(e);
            }
        }
    }

    /**
     * 是否为当前调度服务负责
     */
    @Override
    public boolean isOwner(String name, String uuid) throws Exception {
        boolean isOwner = false;
        // 查看集群中是否注册当前任务，如果没有就自动注册
        String zkPath = this.pathTask + "/" + name;
        if (this.zkManager.isAutoRegisterTask()) {
            if (this.getZooKeeper().exists(zkPath, false) == null) {
                this.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
                if (LOG.isDebugEnabled()) {
                    LOG.debug(uuid + ":自动向集群注册任务[" + name + "]");
                }
                // 重新分配任务
                assignServer2Task(loadScheduleServerNames(), zkPath);
            }
        }
        // 判断是否分配给当前节点
        zkPath = zkPath + "/" + uuid;
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            isOwner = true;
        }
        return isOwner;
    }

    @Override
    public boolean saveRunningInfo(String name, String uuid) throws Exception {
        Date now = new Date();
        // 查看集群中是否注册当前任务，如果没有就自动注册
        String zkPath = this.pathTask + "/" + name;
        // 更新任务执行信息
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            byte[] data = this.getZooKeeper().getData(zkPath, null, null);
            ScheduleTask scheduleTask = null;
            if (null != data) {
                // String json = new String(data);
                String json = (String) ZKTools.deserialize(data);
                scheduleTask = this.gson.fromJson(json, ScheduleTask.class);
                scheduleTask.setRunTimes(scheduleTask.getRunTimes() + 1);
                scheduleTask.setLastRunningTime(now.getTime());
                scheduleTask.setCurrentServer(uuid);
                // this.getZooKeeper().setData(zkPath, this.gson.toJson(scheduleTask).getBytes(), -1);
                byte[] bytes = ZKTools.serialize(this.gson.toJson(scheduleTask));
                this.getZooKeeper().setData(zkPath, bytes, -1);
            }
        }
        // 判断是否分配给当前调度节点
        zkPath = zkPath + "/" + uuid;
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            try {
                int times = 0;
                byte[] dataVal = this.getZooKeeper().getData(zkPath, null, null);
                if (dataVal != null) {
                    // String val = new String(dataVal);
                    String val = (String) ZKTools.deserialize(dataVal);
                    String[] vals = val.split(":");
                    times = Integer.parseInt(vals[0]);
                }
                times++;
                String newVal = times + ":" + now.getTime();
                // this.getZooKeeper().setData(zkPath, newVal.getBytes(), -1);
                byte[] bytes = ZKTools.serialize(newVal);
                this.getZooKeeper().setData(zkPath, bytes, -1);
            } catch (Exception e) {
                // TODO: handle exception
            }
        }
        return true;
    }

    @Override
    public boolean isExistsTask(ScheduleTask scheduleTask) throws Exception {
        String zkPath = this.pathTask + "/" + scheduleTask.getKey();
        return this.getZooKeeper().exists(zkPath, false) != null;
    }

    /**
     * 获取指定调度任务数据
     *
     * @param scheduleKey
     * @return
     * @throws Exception
     * @throws InterruptedException
     * @throws KeeperException
     * @throws JsonSyntaxException
     */
    public ScheduleTask getTask(String scheduleKey) throws Exception {

        ScheduleTask scheduleTask = null;
        String zkPath = this.pathTask;
        zkPath = zkPath + "/" + scheduleKey;
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            byte[] data = this.getZooKeeper().getData(zkPath, null, null);
            // String json = new String(data);
            String json = (String) ZKTools.deserialize(data);
            scheduleTask = this.gson.fromJson(json, ScheduleTask.class);
        }

        return scheduleTask;
    }

    /**
     * 如果不存在，添加； 如果已经存在，则抛出异常。
     */
    @Override
    public void addTask(ScheduleTask scheduleTask) throws Exception {
        String zkPath = this.pathTask;
        zkPath = zkPath + "/" + scheduleTask.getKey();
        if (this.getZooKeeper().exists(zkPath, false) == null) {
            // 创建任务节点
            this.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
            // 保存任务数据
            String json = this.gson.toJson(scheduleTask);
            // this.getZooKeeper().setData(zkPath, json.getBytes(), -1);
            byte[] bytes = ZKTools.serialize(json);
            this.getZooKeeper().setData(zkPath, bytes, -1);
        } else {
            LOG.warn("the task[key: {}] is exists.", scheduleTask.getKey());
            throw new RuntimeException("the task is exists");
        }

    }

    /**
     * 如果更新的任务不存在，则添加； 如果更新的任务的key已经存在，而且任务调度表达式与之前不一致则重新添加调度任务并取消原有调度任务，如果表达式一致则不需要重新调度。
     *
     * .schedule.core.ScheduleTask)
     */
    @Override
    public void createAndUpdateTask(ScheduleTask scheduleTask) throws Exception {
        String zkPath = this.pathTask;
        zkPath = zkPath + "/" + scheduleTask.getKey();
        if (this.isExistsTask(scheduleTask)) {
            ScheduleTask serverTask = this.getTask(scheduleTask.getKey());
            if (!scheduleTask.equals(serverTask)) {
                String json = this.gson.toJson(scheduleTask);
                byte[] bytes = ZKTools.serialize(json);
                this.getZooKeeper().setData(zkPath, bytes, -1);
                // this.getZooKeeper().setData(zkPath, json.getBytes(), -1);
                DynamicTaskManager.cancelTask(scheduleTask.getKey(), false);
            }
        } else {
            this.addTask(scheduleTask);
        }
    }

    @Override
    public void delTask(ScheduleTask scheduleTask) throws Exception {
        String zkPath = this.pathTask;
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            zkPath = zkPath + "/" + scheduleTask.getKey();
            if (this.getZooKeeper().exists(zkPath, false) != null) {
                ZKTools.deleteTree(this.getZooKeeper(), zkPath);
            }
        }
    }

    @Override
    public List<ScheduleTask> selectTask() throws Exception {
        String zkPath = this.pathTask;
        List<ScheduleTask> scheduleTasks = new ArrayList<>();
        if (this.getZooKeeper().exists(zkPath, false) != null) {
            List<String> childes = this.getZooKeeper().getChildren(zkPath, false);
            for (String child : childes) {
                byte[] data = this.getZooKeeper().getData(zkPath + "/" + child, null, null);
                ScheduleTask scheduleTask = null;
                if (null != data) {
                    // String json = new String(data);
                    String json = (String) ZKTools.deserialize(data);
                    scheduleTask = this.gson.fromJson(json, ScheduleTask.class);
                    // scheduleTask.setType("backend task");
                } else if (child.indexOf("#") > 0) {
                    String[] names = child.split("#");
                    if (StringUtils.isNotEmpty(names[0])) {
                        scheduleTask = new ScheduleTask();
                        scheduleTask.setTargetClazz(names[0]);
                        scheduleTask.setTargetMethod(names[1]);
                        scheduleTask.setType("quartz/spring task");
                    }
                } else {
                    scheduleTask = new ScheduleTask();
                    scheduleTask.setKey(child);
                }

                if (scheduleTask.getRunTimes() == 0) { // 如果不是backend Task，则没有记录runtimes和lastRuntime，需要从子节点获取
                    List<String> sers = this.getZooKeeper().getChildren(zkPath + "/" + child, false);
                    if (scheduleTask != null && sers != null && !sers.isEmpty()) {
                        scheduleTask.setCurrentServer(sers.get(0));
                        byte[] dataVal = this.getZooKeeper().getData(zkPath + "/" + child + "/" + sers.get(0), null,
                                null);
                        if (dataVal != null) {
                            // String val = new String(dataVal);
                            String val = (String) ZKTools.deserialize(dataVal);
                            String[] vals = val.split(":");
                            scheduleTask.setRunTimes(Integer.valueOf(vals[0]));
                            scheduleTask.setLastRunningTime(Long.valueOf(vals[1]));
                        }
                    }
                }
                scheduleTasks.add(scheduleTask);
            }
        }
        System.out.println(scheduleTasks.size());
        return scheduleTasks;
    }

    /**
     * 调度当前服务负责的任务
     *
     * @param currentUuid 当前调度服务的uuid
     */
    @Override
    public boolean checkLocalTask(String currentUuid) throws Exception {
        if (this.zkManager.checkZookeeperState()) {
            String zkPath = this.pathTask;
            List<String> children = this.getZooKeeper().getChildren(zkPath, false);
            List<String> ownerTask = new ArrayList<>();
            if (null != children && !children.isEmpty()) {
                for (String taskName : children) {
                    if (isOwner(taskName, currentUuid)) { // 是否为当前调度服务负责
                        String taskPath = zkPath + "/" + taskName;
                        byte[] data = this.getZooKeeper().getData(taskPath, null, null);
                        if (null != data) {
                            // String json = new String(data);
                            String json = (String) ZKTools.deserialize(data);
                            ScheduleTask scheduleTask = this.gson.fromJson(json, ScheduleTask.class);
                            // 判断是否为一次性任务，如果已经执行则不再执行
                            if (scheduleTask.getCronExpression() == null && scheduleTask.getStartTime() != null
                                    && scheduleTask.getPeriod() == 0 && scheduleTask.getRunTimes() > 0) {
                                addTaskHistory(taskName, data, currentUuid);
                                continue;
                            }
                            ownerTask.add(taskName);
                            DynamicTaskManager.scheduleTask(scheduleTask);
                        }
                    }
                }
            }
            // 清理本地多余任务
            DynamicTaskManager.clearLocalTask(ownerTask);
        }
        return false;
    }

    /**
     * @param taskName    当前任务的key值
     * @param data        当前任务的value值(任务详情，并且执行次数已经为1)
     * @param currentUuid 当前调度服务器id
     * @throws Exception
     */
    private void addTaskHistory(String taskName, byte[] data, String currentUuid) throws Exception {
        String taskPath = this.pathTask + "/" + taskName;
        String historyPath = this.pathHistory + "/" + taskName;
        if (this.getZooKeeper().exists(historyPath, false) == null) {
            ZKTools.createPath(getZooKeeper(), historyPath, CreateMode.PERSISTENT, this.zkManager.getAcl());
        }
        this.getZooKeeper().setData(historyPath, data, -1);
        // 当前调度节点信息
        String taskServerPath = taskPath + "/" + currentUuid;
        String historyServerPath = historyPath + "/" + currentUuid;
        // dataVal为任务的执行次数:执行时间戳
        byte[] dataVal = this.getZooKeeper().getData(taskServerPath, null, null);
        if (this.getZooKeeper().exists(historyServerPath, false) == null) {
            ZKTools.createPath(getZooKeeper(), historyServerPath, CreateMode.PERSISTENT, this.zkManager.getAcl());
        }
        this.getZooKeeper().setData(historyServerPath, dataVal, -1);
        if (this.getZooKeeper().exists(taskPath, false) != null) {
            ZKTools.deleteTree(this.getZooKeeper(), taskPath);
        }
    }

}
